package org.example;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.awt.geom.Point2D;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;

import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class PathAggre {
    private static final Logger LOG = LoggerFactory.getLogger(PathAggre.class);

    static class FlinkRabbitMqSource extends RMQSource {
        public String exchangeName;

        public FlinkRabbitMqSource(RMQConnectionConfig rmqConnectionConfig, String queueName,
                                   String exchangeName) {
            super(rmqConnectionConfig, queueName, new SimpleStringSchema(StandardCharsets.UTF_8));
            this.exchangeName = exchangeName;
        }

        @Override
        protected void setupQueue() throws IOException {
            AMQP.Queue.DeclareOk result = channel.queueDeclare(queueName, true, false, false, null);
//            channel.queueBind(result.getQueue(), exchangeName, "*");
        }
    }

    /**
     * flat map input: uuid & corrd list
     * "{\"id\":\"thisisauuid\", \"coarse_path\":
     * [{\"lon\": 121.35191085048595, \"lat\": 18.66431561231426},
     * {\"lon\": 121.30671673629185, \"lat\": 18.678697160768124},
     * ...,
     * {\"lon\": ..., \"lat\": ...}}]}";
     *
     * flat map output:
     * Tuple3(uuid, 0, json"PathFindJobBase"),
     * Tuple3(uuid, 1, json"MultiPointsInOneMap(map id & corrd list [...])"),
     * Tuple3(uuid, 2, json"TwoPointsInTwoMap(map1 map2 & corrd1 coord2)"),
     * ...
     * Tuple3(uuid, n, json"PathFindJobBase")
     *
     * P.S MultiPointsInOneMap & TwoPointsInTwoMap implement based PathFindJobBase.
     */
    static class ParsePointsToJobs implements FlatMapFunction<String, Tuple4<String, Integer, Integer, String>> {
        @Override
        public void flatMap(String jobJson, Collector<Tuple4<String, Integer, Integer, String>> out) throws Exception {
            var o = JSONObject.parseObject(jobJson);
            var uuid = o.getString("uuid");
            var points = o.getJSONArray("coarse_path");
            ArrayList<Point2D.Double> pts = new ArrayList<>();
            int size = points.size();
            for (int i=0; i!=size; i++) {
                var lonLat = points.getJSONObject(i);
                var lon = lonLat.getDouble("lon");
                var lat = lonLat.getDouble("lat");
                pts.add(new Point2D.Double(lon, lat));
            }

            ArrayList<QTreeMapAggre.PathFindJobBase> jobs = QTreeMapAggre.from2dPointArray(pts);
            for (int i=0; i!=jobs.size(); i++) {
                out.collect(Tuple4.of(uuid, i, jobs.size(), jobs.get(i).toJson()));
            }
        }
    }

    static class MakeReturns implements MapFunction<Tuple4<String, Integer, Integer, String>, String> {
        @Override
        public String map(Tuple4<String, Integer, Integer, String> value) throws Exception {
            String uuid = value.f0;
            // {"0":[[x0, y0], [x1, y1], ... [xn, yn]], "1":[[],...], "2":[[], ...] }
            String pointList = value.f3;
            JSONObject o = JSON.parseObject(pointList);
            ArrayList<String> arr = new ArrayList<>(o.keySet());
            arr.sort((str1, str2)->{
                Integer i1 = Integer.parseInt(str1);
                Integer i2 = Integer.parseInt(str2);
                return i1.compareTo(i2);
            });

            JSONArray jr = new JSONArray();
            for (var it : arr) {
                var pts = o.getJSONArray(it);  // [[],[]]
                for (int i = 0; i!=pts.size(); i++) {
                    var p = pts.getJSONArray(i);
                    var lonOffset = p.getIntValue(0)/100.0;
                    var latOffset = p.getIntValue(1)/100.0;
                    JSONObject r = new JSONObject();
                    r.put("lon", QTreeMapAggre.lonBase+lonOffset);
                    r.put("lat", QTreeMapAggre.latBase+latOffset);
                    jr.add(r);
                }
            }

            JSONObject retO = new JSONObject();
            retO.put("uuid", uuid);
            retO.put("point_detail", jr);
            LOG.info("[****]Job return value made[****]");
            return retO.toJSONString();
        }
    }

    /**
     * input data:
     * Tuple3(uuid, index, JSONSTR"PathFindJobBase map ... "),
     * Tuple3(uuid, index, JSONSTR"MultiPointsInOneMap: map id & corrd list [...]"),
     * Tuple3(uuid, index, JSONSTR"TwoPointsInTwoMap: map1 map2 & corrd1 coord2"),
     *
     * output:
     * Tuple3(uuid, index, map id & routes"),
     * Tuple3(uuid, index, map id & routes"),
     */
    static class CalcRoute implements MapFunction<Tuple4<String, Integer, Integer, String>, Tuple4<String, Integer, Integer, String>> {
        @Override
        public Tuple4<String, Integer, Integer, String> map(Tuple4<String, Integer, Integer, String> jobTuple) throws Exception {
            String uuid = jobTuple.f0;
            Integer index = jobTuple.f1;
            Integer total = jobTuple.f2;
            String jobJsonStr = jobTuple.f3;


            QTreeMapAggre.PathFindJobBase job = QTreeMapAggre.jobFromJson(jobJsonStr);
            long timeMillis = System.currentTimeMillis();
            ArrayList<Point> rs = new ArrayList<>();
            if (job.getClass() == QTreeMapAggre.MultiPointsInOneMap.class) {
                QTreeMapAggre.MultiPointsInOneMap s = (QTreeMapAggre.MultiPointsInOneMap) job;
                if (s.points.size() < 2) {
                    return Tuple4.of(uuid, index, total, "[]");
                }
                var seri = new QTreeSerializer();
                String fp = "/var/flinkrc/pytile/" + s.mapBase.mx + "_" + s.mapBase.my + ".qt";
                var quadTree = seri.deserializeCustom(fp);
                var afg = new QTreeFromGrid();
                afg.computeAllNeighbors(quadTree);
                //
                ArrayList<QTreeMapAggre.Offset[]> v = new ArrayList<>();
                for (int i = 0; i != s.points.size() - 1; i++) {
                    var prev = new QTreeMapAggre.Offset((s.points).get(i).ox, (s.points).get(i).oy);
                    var next = new QTreeMapAggre.Offset((s.points).get(i + 1).ox, (s.points).get(i + 1).oy);
                    QTreeMapAggre.Offset[] pair = new QTreeMapAggre.Offset[]{prev, next};
                    v.add(pair);
                }
                v.forEach(w -> {
                    var route = new AStar(quadTree).findPath(new Point(w[0].ox, w[0].oy), new Point(w[1].ox, w[1].oy));
                    if (route.isEmpty()) {
                        LOG.error("An empty path was found, from ({},{}) to ({},{})", w[0].ox, w[0].oy, w[1].ox, w[1].oy);
                        LOG.error("A direct line was made to prevent empty route.");
                        var u = getLinePointsDDA(
                                s.mapBase.mx + w[0].ox, s.mapBase.my + w[0].oy,
                                s.mapBase.mx + w[1].ox, s.mapBase.my + w[1].oy);
                        rs.addAll(u);
                    } else {
                        ArrayList<Point> route2 = new ArrayList<>();
                        route.forEach(a -> {
                            route2.add(new Point(s.mapBase.mx + a.x, s.mapBase.my + a.y));
                        });
                        rs.addAll(route2);
                    }
                });
            } else {
                QTreeMapAggre.TwoPointsInTwoMap s = (QTreeMapAggre.TwoPointsInTwoMap) job;
                var map1 = s.mapBase1;
                var map2 = s.mapBase2;
                var pt1 = s.offset1;
                var pt2 = s.offset2;
                var u = getLinePointsDDA(map1.mx + pt1.ox, map1.my + pt1.oy, map2.mx + pt2.ox, map2.my + pt2.oy);
                LOG.info("still not know how to find a path between two maps, draw a line instead temporally.");
                rs.addAll(u);
            }
            LOG.info("[****]job {} mainly work is finished in {} ms![****]", uuid, System.currentTimeMillis() - timeMillis);
            var r = new JSONArray();
            for (var it : rs) {
                var u = new JSONArray();
                u.add(it.x);
                u.add(it.y);
                r.add(u);
            }
            var o = new JSONObject();
            o.put(index.toString(), r);   // {"1": [[],[],...]}
            return Tuple4.of(uuid, 1, total, o.toJSONString());
        }

        public static ArrayList<Point> getLinePointsDDA(int x0, int y0, int x1, int y1) {
            ArrayList<Point> points = new ArrayList<>();

            int dx = x1 - x0;
            int dy = y1 - y0;
            int steps = Math.max(Math.abs(dx), Math.abs(dy));

            if (steps == 0) {
                points.add(new Point(x0, y0));
                return points;
            }

            double xIncrement = (double) dx / steps;
            double yIncrement = (double) dy / steps;

            double x = x0;
            double y = y0;

            for (int i = 0; i <= steps; i++) {
                points.add(new Point((int) Math.round(x), (int) Math.round(y)));
                x += xIncrement;
                y += yIncrement;
            }
            return points;
        }
    }

    static class MyReduceFunction implements ReduceFunction<Tuple4<String, Integer, Integer, String> > {
        public Tuple4<String, Integer, Integer, String>  reduce(
                Tuple4<String, Integer, Integer, String> value1,
                Tuple4<String, Integer, Integer, String> value2) throws Exception
        {
            // uuid, index, total, json str
            JSONObject o1 = JSON.parseObject(value1.f3);
            JSONObject o2 = JSON.parseObject(value2.f3);
            JSONObject o3 = o1;
            o2.keySet().forEach(s->{
                var arr = o2.getJSONArray(s);
                o3.put(s, arr);
            });
            return Tuple4.of(value1.f0, value2.f1 + value1.f1, value1.f2, o3.toJSONString());
        }
    }

    static class MyFilterFunction implements FilterFunction<Tuple4<String, Integer, Integer, String> > {
        @Override
        public boolean filter(Tuple4<String, Integer, Integer, String> value) throws Exception {
            return value.f1.equals(value.f2);
        }
    }

    public static void main(String[] args) throws Exception {
        System.out.println("hello, main");
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost("127.0.0.1")
                .setPort(5672)
                .setUserName("duoduo")
                .setPassword("123456")
                .setVirtualHost("/")
                .build();
        var mqSource = new FlinkRabbitMqSource(connectionConfig,
                "queue_name1",
                "");

        DataStreamSource dataStreamSource = env.addSource(mqSource);
        var e = dataStreamSource.flatMap(new ParsePointsToJobs())
                .map(new PathAggre.CalcRoute())
                .keyBy(0)
                .reduce(new MyReduceFunction())
                .filter(new MyFilterFunction()).map(new MakeReturns());
        e.print();

        final RMQConnectionConfig sinkConnConfig = new RMQConnectionConfig.Builder()
                .setHost("127.0.0.1")
                .setPort(5672)
                .setUserName("guest")
                .setPassword("guest")
                .setVirtualHost("/")
                .build();
        e.addSink(new RMQSink<String>(
                connectionConfig,            // config for the RabbitMQ connection
                "test_queue",                 // name of the RabbitMQ queue to send messages to
                new SimpleStringSchema()));  // serialization schema to turn Java objects to messages
//
//        var f = e.filter(new MyFilterFunction());
//        f.print();
//
//        var g = f.map(new MakeReturns());
//        g.print();

        // e.print();
        //e.sinkTo();
        //var x = dataStreamSource.map(new PathAggre.Test3MapFunction())
        //x.print()
        //FileSink<String> sink = FileSink
        //        .forRowFormat(new Path("./output/path"), new SimpleStringEncoder<String>("UTF-8"))
        //        .build();
        //x.sinkTo(sink);
//        var mergedMapPoints = QTreeMapAggre.getJobMapAndPoints();
//        ArrayList<Tuple2<Integer, String>> s = new ArrayList<>();
//        for (int i=0; i!=mergedMapPoints.size(); i++) {
//            s.add(Tuple2.of(i, mergedMapPoints.get(i).toJson()));
//        }
//        DataStreamSource<Tuple2<Integer, String>> r = env.fromCollection(s);
//        var x = r.map(new PathAggre.Test2MapFunction());

        env.execute("Flink Java API Skeleton");
    }


}
